package com.leilei;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/9 22:34
 * @desc flink数据源合并 connect特性
 * connect 最多只能合并两个数据流，且合并后无法直接sink,需要进行数据处理
 */
public class ConnectOperator {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        env.setParallelism(1);
        DataStreamSource<String> source = env.fromElements("zs", "li", "we");
        DataStreamSource<Integer> source2 = env.fromElements(1, 2, 3, 4, 5, 6, 7);
        ConnectedStreams<String, Integer> connect = source.connect(source2);
        SingleOutputStreamOperator<String> streamOperator = connect.map(new CoMapFunction<String, Integer, String>() {
            @Override
            public String map1(String value) {
                return value + "是字符串类型,直接加后缀";
            }

            @Override
            public String map2(Integer value) {
                return "原本是Integer类型:" + value + "现在也变为String";
            }
        });
        streamOperator.print("connect");
        env.execute();
    }
}
